#ifndef _PUBSUB_H
#define _PUBSUB_H

#include "broker_protocol.pb.h"
#include "transport.h"
#include <string>
#include <set>

// One important principle about subclassing the subscriber is:
// AVOID making on_message() a blocking call.
// A blocking on_message() works, but it has deadly drawbacks.
//   1) it may delay the execution of other subscribers.
//   2) in the typical usage scenario, subscribers often needs to hold
//      a mutex while handling an incoming message. If on_message() blocks, the
//      subscriber won't be able to process further incoming data.
// To carry out this principle, we have to at least follow two rules:
//   1) never make calls to blocking functions, e.g., blocking recv()
//   2) never wait on a particular conditional variable in the handler function

class subscriber {
public:
    virtual void on_message(const DataMessage *msg) = 0;
};

// a convenient wrapper for subscribers on file descriptors
class fd_subscriber : public subscriber {
public:
    fd_subscriber(int);
    ~fd_subscriber();

    void on_message(const DataMessage *msg);

private:
    int fd_;
};

/**
* A subscription-based messaging core implemented upon the transport layer.
* Other parties use this module to publish content and to receive subscribed
* content from the outside world.
* XXX pattern-based subscription
*/
class messaging : public transport_listener {
public:
    messaging(const std::string &myid, const std::vector<std::string> &options);
    ~messaging();

    /**
     * Process a message. This message could be from a plugin or a function
     * within the same process.
     */
    void deliver(DataMessage *msg);

    void subscribe(const std::string &subject, subscriber *s);
    void unsubscribe(const std::string &subject, subscriber *s);

    std::string id() const;

private:
    std::string id_;
    transport *trans_;
    pthread_mutex_t m_;
    std::map<std::string, std::set<subscriber *> > sub_;
    std::map<std::string, pthread_mutex_t *> locks_;

    void on_data(const std::string &src, const std::string &data);
    void notify(const DataMessage *msg);
};

#endif
